package org.morning.open.connectionPool;

import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 通用网络连接池
 * <p>
 * 目前自动关闭多余的空闲连接没实现
 * 
 * @author haigege
 * @date 2017-12-06
 * @version 0.0.1
 */
public class ConnectionPool<T> {
	org.slf4j.Logger logger = org.slf4j.LoggerFactory
			.getLogger(ConnectionPool.class);
	/**
	 * 地址
	 */
	private String host = null;
	/**
	 * 端口
	 */
	private int port = 0;
	/**
	 * 创建连接需要的额外参数
	 */
	private Map<String, Object> args = null;
	/**
	 * 空闲连接
	 */
	private LinkedBlockingDeque<T> idleConnections = null;
	/** 等待线程队列，先进先出 */
	private final LinkedList<Object> waitQueue = new LinkedList<Object>();
	/** 最小连接数 */
	private int minSize = 1;
	/** 最大连接数 */
	private int maxSize = 1;
	/** 最大空闲链接数 */
	private int maxIdle = 1;
	/** 连接池中的总连接数 */
	private int totalSize = 0;
	/** 连接池调整大小 */
	private int step = 1;
	/**
	 * 改变连接池时的锁（扩容或者缩容）
	 */
	private final ReentrantLock changePooLock = new ReentrantLock();
	/**
	 * 存储所有连接的集合
	 */
	private LinkedHashSet<T> connectionSet = null;
	/** 连接创建器 */
	private ConnectionBuilder<T> connectionBuilder = null;
	/**
	 * 线程池，用于异步扩容、缩容、关闭连接
	 */
	private final ExecutorService threadPool = Executors
			.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
	/** 连接池是否已经初始化 */
	private boolean initialized = false;

	private void base(String host, int port,
			ConnectionBuilder<T> connectionBuilder) {
		this.host = host;
		this.port = port;
		if (connectionBuilder == null) {
			throw new IllegalArgumentException(
					"connectionBuilder should not be null");
		}
		this.connectionBuilder = connectionBuilder;
	}

	/**
	 * 
	 * @param host
	 *            网络地址
	 * @param port
	 *            端口
	 * @param connectionBuilder
	 *            网络连接创建器（需要根据自己的通讯协议实现创建器接口）
	 */
	public ConnectionPool(String host, int port,
			ConnectionBuilder<T> connectionBuilder) {
		base(host, port, connectionBuilder);
		initPool();
	}

	/**
	 * 
	 * @param host
	 *            网络地址
	 * @param port
	 *            端口
	 * @param connectionBuilder
	 *            网络连接创建器（需要根据自己的通讯协议实现创建器接口）
	 * @param args
	 *            创建网络连接需要的附加参数
	 */
	public ConnectionPool(String host, int port,
			ConnectionBuilder<T> connectionBuilder, Map<String, Object> args) {
		base(host, port, connectionBuilder);
		this.args = args;
		initPool();
	}

	/**
	 * 
	 * @param host
	 *            网络地址
	 * @param port
	 *            端口
	 * @param connectionBuilder
	 *            网络连接创建器（需要根据自己的通讯协议实现创建器接口）
	 * @param minSize
	 *            最小链接数（初始化链接数）
	 * @param maxSize
	 *            最大链接数
	 * @param maxIdle
	 *            最大空闲链接数
	 * @param args
	 *            创建网络连接需要的附加参数
	 */
	public ConnectionPool(String host, int port,
			ConnectionBuilder<T> connectionBuilder, int minSize, int maxSize,
			int maxIdle, Map<String, Object> args) {
		base(host, port, connectionBuilder);
		this.args = args;
		this.minSize = minSize;
		this.maxSize = maxSize;
		this.maxIdle = maxIdle;
		initPool();
	}

	/**
	 * 
	 * @param host
	 *            网络地址
	 * @param port
	 *            端口
	 * @param connectionBuilder
	 *            网络连接创建器（需要根据自己的通讯协议实现创建器接口）
	 * @param minSize
	 *            最小链接数（初始化链接数）
	 * @param maxSize
	 *            最大链接数
	 * @param maxIdle
	 *            最大空闲链接数
	 */
	public ConnectionPool(String host, int port,
			ConnectionBuilder<T> connectionBuilder, int minSize, int maxSize,
			int maxIdle) {
		base(host, port, connectionBuilder);
		this.minSize = minSize;
		this.maxSize = maxSize;
		this.maxIdle = maxIdle;
		initPool();
	}

	/**
	 * 初始化池
	 */
	private synchronized void initPool() {
		if (initialized) {
			return;
		}
		initialized = true;
		logger.debug("connection pool initialized!");

		// 初始化属性，基础数据结构
		this.step = this.maxIdle;
		this.idleConnections = new LinkedBlockingDeque<T>(maxSize);
		this.connectionSet = new LinkedHashSet<T>(maxSize);

		this.changePooLock.lock();
		try {
			for (int i = 0; i < minSize; i++) {
				try {
					T conn = createConnection(host, port, args);
					if (conn != null) {
						idleConnections.add(conn);
						++totalSize;
						logger.trace("increase a connection, "
								+ "total connections = " + totalSize
								+ ", idle connections = "
								+ idleConnections.size());
					}
				} catch (Exception e) {// 初始化过程异常，需要外层程序处理，决定是否继续
					throw new RuntimeException("init connection pool error", e);
				}
			}
		} finally {
			this.changePooLock.unlock();
		}
	}

	/**
	 * 获取连接
	 */
	public T getConnection() {
		T result = null;
		int waitCount = 0;
		while (true) {// 直到获取到一条连接为止
			result = internalGetConnection(waitCount);

			if (result != null) {
				logger.debug("thread " + Thread.currentThread().getName()
						+ " aquired a connection, " + "total connections = "
						+ totalSize + ", idle connections = "
						+ idleConnections.size());
				break;
			} else {
				Object monitor = new Object();
				logger.debug("thread " + Thread.currentThread().getName()
						+ " wait for a connection.");

				// 没有获取到连接，将当前线程加入等待队列
				synchronized (monitor) {
					synchronized (waitQueue) {
						waitQueue.add(monitor);
					}
					try {
						waitCount++;
						monitor.wait();
					} catch (InterruptedException ignore) {
					}
					// 唤醒后会继续回到while循环，尝试获取连接
				}

				logger.debug("thread " + Thread.currentThread().getName()
						+ " wakeup.");
			}
		}

		return result;

	}

	/**
	 * 获取连接，如果没有连接，则通知增加连接池
	 */
	private T internalGetConnection(int waitCount) {
		T result = null;
		if (idleConnections.isEmpty()) {// 如果已经没有空闲连接了
			logger.debug("current pool is empty, "
					+ "try to increase connection pool.");
			// 通知做一次增加池内连接数量操作
			notifyIncreasePool();
			if (waitCount == 0) {// 并且如果获取连接的线程第一次获取,未曾等待,直接回去排队公平竞争下一波空闲连接（保持大部分人的公平）
				return null;// 直接回去排队公平竞争下一波空闲连接
			} else {// 否则同步增加一个连接,当然有可能拿不到,那就进到下面小房间去
				try {
					T conn = createConnection(host, port, args);
					if (conn != null) {//同步获得一个连接	
						logger.trace("increase(Blocking) a connection, " + "total connections = "
								+ totalSize + ", idle connections = "
								+ idleConnections.size());
						changePooLock.lock();
						try {
							totalSize++;							
							return conn;
						} finally{
							changePooLock.unlock();							
						}						
					}
				} catch (Exception e) {
					logger.warn("create connection error", e);
				}
			}
		}
		// 加入进来参与激烈的不公平（不排队）竞争（总是，原则就是屋子外面的人进来要排队，屋子里面的人（不能太多）拿到资源靠的是勇气和运气）
		try {
			result = idleConnections.take();// 这里如果多个线程（数量不会多）在空闲连接取完之前同时挤进来，那么参与不公平竞争（保证这小波人快速拿到可用连接）
		} catch (InterruptedException e) {
			logger.error("take out an idle connection error", e);
		}
		if (idleConnections.isEmpty()) {// 可用链接数已空
			if (totalSize < maxSize) {// 当前创建的连接总数小于最大连接数，通知增加连接池
				logger.debug("current pool is empty, "
						+ "try to increase connection pool.");
				notifyIncreasePool();
			}
		}
		return result;
	}

	/**
	 * 通知对连接池做一次扩容
	 */
	private void notifyIncreasePool() {
		// 这里需要考虑，从正向反馈通知扩容，到扩容完成负向反馈停止继续扩容需要一段时间，如果没有弹簧，很容易吃撑
		threadPool.execute(new Runnable() {

			@Override
			public void run() {
				boolean todo = false;
				int localStep = 1;// 默认步长
				changePooLock.lock();
				try {
					if (totalSize < maxSize) {
						if (idleConnections.size() > maxIdle * 3) {// 因为扩容过程有延迟，如果准备扩容之前发现空闲连接很多，那么没有必要扩容
							;
						} else {
							if (totalSize + step < maxSize) {// 只要没有达到最大链接数，既然通知扩容，就按配置步长做一次扩容
								localStep = step;
							}
							totalSize = totalSize + localStep;// 预扩容，如果实际扩容过程发生错误，没有增加到这些连接，再扣掉（这样做是为了提高多线程扩容的效率）
							todo = true;
						}
					}

				} finally {
					changePooLock.unlock();
				}
				if (todo) {// 这里提前释放掉锁,开始实际扩容过程（可以并发扩容）
					int rollbackNum = 0;
					while (localStep-- > 0) {
						if (!addOneIdleConnection()) {// 如果增加连接失败
							rollbackNum++;
						}
					}
					if (rollbackNum > 0) {
						changePooLock.lock();
						try {
							totalSize = totalSize - rollbackNum;
						} finally {
							changePooLock.unlock();
						}
					}
				}
				// 如果存在空闲可用连接，唤醒排队等待线程
				if (idleConnections.size() > 0) {
					notifyWaitingThreads(idleConnections.size());
				}
			}
		});
	}

	/**
	 * 新增一个空闲连接
	 * 
	 * @return
	 */
	private boolean addOneIdleConnection() {
		T conn = null;
		try {
			conn = createConnection(host, port, args);
		} catch (Exception e) {
			logger.warn("create connection error", e);
		}
		if (conn != null) {
			idleConnections.add(conn);// 成功增加一个连接
			logger.trace("increase a connection, " + "total connections = "
					+ totalSize + ", idle connections = "
					+ idleConnections.size());
			return true;
		}
		return false;
	}

	/**
	 * 创建网络连接
	 * 
	 * @param host
	 * @param port
	 * @param args
	 * @return
	 */
	private T createConnection(String host, int port, Map<String, Object> args) {
		T conn = connectionBuilder.create(host, port, args);
		if (conn != null) {// 加入到连接集合里
			synchronized (connectionSet) {
				connectionSet.add(conn);
			}
		}
		return conn;
	}

	/**
	 * 释放连接
	 * 
	 * @param connection
	 */
	public void releaseConnection(T connection) {
		idleConnections.add(connection);
		logger.debug("release a connection, " + "total connections = "
				+ totalSize + ", idle connections = " + idleConnections.size());

		notifyWaitingThreads(1);
	}

	/**
	 * 释放坏掉的连接
	 * 
	 * @param connection
	 */
	public void releaseBrokenConnection(T connection) {
		asynchronizedCloseConnection(connection);// 异步关闭坏掉的连接
	}

	/**
	 * 带句柄的异步关闭连接方法
	 * 
	 * @param connection
	 * @return 成功关闭返回feFuture.get()==null, 否则feFuture.get()==connection
	 */
	private Future<T> asynchronizedCloseConnection(final T connection) {
		return this.threadPool.submit(new Callable<T>() {

			@Override
			public T call() throws Exception {
				try {
					connectionBuilder.close(connection);
					synchronized (connectionSet) {
						if(connectionSet.remove(connection)){
							changePooLock.lock();
							try {
								totalSize--;
							} finally{
								changePooLock.unlock();
							}							
						}
					}
					logger.debug("close a connection, "
							+ "total connections = " + totalSize
							+ ", idle connections = " + idleConnections.size());
				} catch (Exception e) {
					logger.warn("close a connection error", e);
					return connection;
				}
				return null;
			}
		});
	}

	/**
	 * 唤醒等待的线程
	 */
	private void notifyWaitingThreads(int num) {
		for (int i = 0; i < num; i++) {
			Object waitMonitor = null;
			synchronized (waitQueue) {
				if (waitQueue.size() > 0) {
					waitMonitor = waitQueue.removeFirst();
				}
			}
			if (waitMonitor != null) {
				synchronized (waitMonitor) {
					waitMonitor.notify();
				}
			}
		}
	}

	@Override
	protected void finalize() throws Throwable {
		// 释放掉所有连接(尽最大努力,防止一些服务器不会自动释放客户端异常断开的连接)
		synchronized (connectionSet) {
			for (T t : connectionSet) {
				try {
					connectionBuilder.close(t);
					connectionSet.remove(t);
					changePooLock.lock();
					try {
						totalSize--;
					} finally{
						changePooLock.unlock();
					}	
					logger.debug("close a connection, "
							+ "total connections = " + totalSize
							+ ", idle connections = " + idleConnections.size());
				} catch (Exception e) {
					logger.warn("close a connection error", e);
				}
			}
		}
		// 关闭线程池
		try {
			this.threadPool.shutdown();
			while (!this.threadPool.isTerminated()) {
				Thread.sleep(2);
			}
		} catch (Exception e) {
			logger.warn("error occurred in finalize connection pool", e);
		}
	}

}
